//package StrConcate;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.StringTokenizer;
import java.util.regex.Pattern;
import java.util.regex.Matcher;
import java.util.Vector;
import java.util.HashSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class StrConcate extends Configured implements Tool {
  public static class MapClass extends MapReduceBase
    implements Mapper<LongWritable, Text, Text, Text> {
    
    public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter) 
            throws IOException {
      String[] tokens = value.toString().split("\t");
      Text keyToken = new Text(tokens[0]);
      for(int i=1;i<tokens.length;i++) {
          output.collect(keyToken, new Text(tokens[i]));
      }   
    }
  }

  public static class Reduce extends MapReduceBase
    implements Reducer<Text, Text, Text, Text> {
    public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter) 
              throws IOException {
      StringBuilder sb = new StringBuilder();
      while(values.hasNext()) {
          Text value = values.next();
          sb.append(value+",");
      }
      output.collect(key, new Text(sb.toString()));
    }
  }
  
  public int run(String[] args) throws Exception {
    JobConf conf = new JobConf(getConf(), StrConcate.class);
    conf.setJobName("StrConcate");

    conf.setMapperClass(MapClass.class);
    conf.setCombinerClass(Reduce.class);
    conf.setReducerClass(Reduce.class);

    FileInputFormat.setInputPaths(conf, (String)args[0]);
    FileOutputFormat.setOutputPath(conf, new Path((String)args[1]));

    conf.setInputFormat(TextInputFormat.class);
    conf.setOutputFormat(TextOutputFormat.class);

    conf.setOutputKeyClass(Text.class);
    conf.setOutputValueClass(Text.class);
    
    conf.setNumMapTasks(15);
    conf.setNumReduceTasks(1);
    
    JobClient.runJob(conf);
    return 0;
  }
  public static void main(String[] args) throws Exception {
    int res = ToolRunner.run(new Configuration(), new StrConcate(), args);
    System.exit(res);
  }
}